package cn.lecosa.es;

import org.apache.log4j.Level
import org.apache.log4j.Logger
import org.apache.spark.SparkConf
import org.apache.spark.SparkConf
import org.apache.spark.SparkConf
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD._
import org.apache.spark.rdd.RDD.rddToPairRDDFunctions
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.SQLContext._
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.hive.HiveContext
import org.elasticsearch.spark._
import org.elasticsearch.spark.rdd.EsSpark
import org.elasticsearch.spark.sql
import org.elasticsearch.spark.sql._
import org.slf4j.LoggerFactory

/*
* 启动：spark-submit --class com.mininglamp.mesense.HiveToEs --master $master  --jars $cpath mesense_sparkstreaming.jar devices.conf
* 运行脚本：sh hivetoes.sh local
* */
object HiveToEs {
  val logger = LoggerFactory.getLogger(this.getClass)
  // 过滤日志
  Logger.getLogger("org").setLevel(Level.WARN)

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("DecisionTree1").setMaster("local[2]")
    sparkConf.set("es.index.auto.create", "true")
    sparkConf.set("es.nodes", "park01")
    println("hello world");
    sparkConf.set("es.port", "9200")
    val sc = new SparkContext(sparkConf)
    val hiveContext = new HiveContext(sc);
    hiveContext.sql("use lecosa")
    val df = hiveContext.sql("select name,age,salary from emp")
    //    val dataList= new ListBuffer[String]
    df.show(300)
    //    val mapRDD=df.rdd.map(t=>{
    //     val s=if("".equals(t(2)))  null else "{\"name\":\""+t(0)+"\",\"age\":\""+t(1)+"\",\"time\":"+DateUtils.parseLong(t(2).toString)+"}"
    //     s
    //   }
    //   )
    import org.elasticsearch.spark._
    df.saveToEs("spark/emp")
    //    mapRDD.filter(_!=null).saveJsonToEs("userveh/idmerge")
  }
}
